package cn.spark.study.sql

import org.apache.spark.sql.expressions.UserDefinedAggregateFunction
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.types.DataType
import org.apache.spark.sql.types.StructField
import org.apache.spark.sql.types.StringType
import org.apache.spark.sql.types.IntegerType


class StringCount extends UserDefinedAggregateFunction{
  
  //输入参数的类型
    def inputSchema: org.apache.spark.sql.types.StructType = {
    StructType(Array(StructField("name",StringType,true)))
  }
  
  //中间进行聚合时所处理的类型
  def bufferSchema: StructType = {
    StructType(Array(StructField("count",IntegerType,true)))
  }
  
  //函数返回值的类型
  def dataType: DataType = {
    IntegerType
  }
  
  //为每个分组数据执行初始化操作
   def initialize(buffer: org.apache.spark.sql.expressions.MutableAggregationBuffer): Unit = {
    buffer(0) = 0
  }
  
   //每个分组，如果有新的值过来的时候，如何进行分组对应的聚合脂的计算 buffer:之前缓存的值   input：传进来的值
  def update(buffer: org.apache.spark.sql.expressions.MutableAggregationBuffer,input: org.apache.spark.sql.Row): Unit = {
    buffer(0) = buffer.getAs[Int](0) + 1
  }
  
  //由于spark是分布式的，所以一个分组的数据，可能会在不同节点上进行局部聚合，就是update。 但是最后一个分组在各个节点的聚合脂要进行merge也就是合并
  def merge(buffer1: org.apache.spark.sql.expressions.MutableAggregationBuffer,buffer2: org.apache.spark.sql.Row): Unit = {
    buffer1(0) = buffer1.getAs[Int](0) + buffer2.getAs[Int](0)
  }
  
  def deterministic: Boolean = {
    true
  }
  
  //一个分组的集合值，如何通过中间的缓存聚合值，最后返回一个最终的聚合值
  def evaluate(buffer: org.apache.spark.sql.Row): Any = {
    buffer.getAs[Int](0)
  }


}